/**
 * 
 */
package com.cqcis.uip.base.framework.utils;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.mina.common.IoSession;
import org.apache.mina.common.WriteFuture;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.TaskWaitTimeConsumePool;

/**
 * 针对Task的工具类
 * 
 * @author huangmy
 *
 */
public class TaskUtil {	
	private final static String TASK_CTQ_ATTR = TaskUtil.class.getName() + "_TASK_CTQ_ATTR_";
	private final static String TASK_SERVICE_CTQ_ATTR = TaskUtil.class.getName() + "_TASK_SERVICE_CTQ_ATTR";
	private final static String TASK_WAIT_TIME_POOL = TaskUtil.class.getName() + "_TASK_WAIT_TIME_POOL";
	private final static String TASK_WAIT_TIME_WAIT_STATE = TaskUtil.class.getName() + "_TASK_WAIT_TIME_WAIT_STATE";
	private final static String TASK_WAIT_DIRECT_LOCK = TaskUtil.class.getName() + "_TASK_WAIT_DIRECT_LOCK";
	private final static String TASK_WAIT_PENDING = TaskUtil.class.getName() + "_TASK_WAIT_PENDING";
	private final static String TASK_CURRENT_SESSION = TaskUtil.class.getName() + "_TASK_CURRENT_SESSION";
	private final static String TASK_STATE_INVALID = TaskUtil.class.getName() + "_TASK_STATE_INVALID";
	private final static String TASK_COUNTER = TaskUtil.class.getName() + "_TASK_COUNTER_";
	private final static String FISISH_AGAIN_TASK = TaskUtil.class.getName() + "_FISISH_AGAIN_TASK";
	private final static String TASK_HAS_EXCEPTION_RESULT = TaskUtil.class.getName() + "_TASK_HAS_EXCEPTION_RESULT";
	private final static String TASK_HAS_RESULT = TaskUtil.class.getName() + "_TASK_HAS_RESULT";
	private final static String TASK_HAS_FINISHED = TaskUtil.class.getName() + "_TASK_HAS_FINISHED";
	private final static String TASK_TIMEOUT = TaskUtil.class.getName() + "_TASK_TIMEOUT";
	
	/**
	 * 清理
	 * 
	 * @param task
	 */
	public static void clearAttrs(Task task) {
		task.removeAttribute(TASK_CTQ_ATTR);
		task.removeAttribute(TASK_SERVICE_CTQ_ATTR);
		task.removeAttribute(TASK_WAIT_TIME_POOL);
		task.removeAttribute(TASK_WAIT_TIME_WAIT_STATE);
		task.removeAttribute(TASK_WAIT_DIRECT_LOCK);
		task.removeAttribute(TASK_WAIT_PENDING);
		task.removeAttribute(TASK_CURRENT_SESSION);
		task.removeAttribute(TASK_STATE_INVALID);
		task.removeAttribute(TASK_COUNTER);
		task.removeAttribute(FISISH_AGAIN_TASK);
		task.removeAttribute(TASK_HAS_EXCEPTION_RESULT);
		task.removeAttribute(TASK_HAS_RESULT);
		task.removeAttribute(TASK_HAS_FINISHED);
		task.removeAttribute(TASK_TIMEOUT);
	}
	
	/**
	 * 超时时间
	 * 
	 * @param task
	 * @return
	 */
	public static long getTimeout(Task task) {
		Long timeout = (Long)task.getAttribute(TASK_TIMEOUT);
		if (timeout != null) {
			return timeout;
		}
		
		return 0;
	}
	
	/**
	 * 设置超时时间
	 * 
	 * @param task
	 * @param timeout
	 */
	public static void setTimeout(Task task, long timeout) {
		task.setAttribute(TASK_TIMEOUT, Long.valueOf(timeout));
	}
	
	/**
	 * 已经生成异常结果
	 * 
	 * @param task
	 */
	public static void setHasGenerateExceptionResult(Task task) {
		task.setAttribute(TASK_HAS_EXCEPTION_RESULT, "true");
	}
	
	/**
	 * 删除
	 * 
	 * @param task
	 */
	public static void removeHasGenerateExceptionResult(Task task) {
		task.removeAttribute(TASK_HAS_EXCEPTION_RESULT);
	}
	
	/**
	 * 是否已经生成异常结果
	 * 
	 * @param task
	 * @return
	 */
	public static boolean isHasGenerateExceptionResult(Task task) {
		return task.getAttribute(TASK_HAS_EXCEPTION_RESULT) != null ? true : false;
	}
	
	/**
	 * 已经生成结果
	 * 
	 * @param task
	 */
	public static void setHasGenerateResult(Task task) {
		task.setAttribute(TASK_HAS_RESULT, "true");
	}
	
	/**
	 * 是否已经生成结果
	 * 
	 * @param task
	 * @return
	 */
	public static boolean isHasGenerateResult(Task task) {
		return task.getAttribute(TASK_HAS_RESULT) != null ? true : false;
	}
	
	/**
	 * 已经生成结果
	 * 
	 * @param task
	 */
	public static void setTaskHasGenerateFinished(Task task) {
		task.setAttribute(TASK_HAS_FINISHED, "true");
	}
	
	/**
	 * 是否已经生成结果
	 * 
	 * @param task
	 * @return
	 */
	public static boolean isTaskHasGenerateFinished(Task task) {
		return task.getAttribute(TASK_HAS_FINISHED) != null ? true : false;
	}
	
	/**
	 * 设置完成的again Task标记
	 * 
	 * @param task
	 */
	public static void setFinishedAgainTaskFlag(Task task) {
		task.setAttribute(FISISH_AGAIN_TASK, "true"); 
	}
	
	/**
	 * 是否完成的again Task
	 * 
	 * @param task
	 * @return
	 */
	public static boolean isFinishedAgainTask(Task task) {
		return task.getAttribute(FISISH_AGAIN_TASK) != null ? true : false;
	}
	
	/**
     * 是否同步Task
     * 
     * @param task
     * @return
     */
    public static boolean isSyncTask(Task task) {
    	return task.getAttribute(FrameworkConstants.TASK_ATTR_SYNC) != null ? true : false;  	
    }
    
    /**
     * 设置同步
     * 
     * @param task
     */
    public static void setSyncTask(Task task) {
    	task.setAttribute(FrameworkConstants.TASK_ATTR_SYNC, "true"); // 同步任务
    }
    
    /**
     * 发送对象，异步
     * 
     * @param task
     * @param o
     */
    public static void write(Task task, Object o) {
    	synchronized (task) {
			IoSession session = getCurrentSession(task);
			if (session != null) {
				session.write(o);
			}
		}
    }
    
    /**
     * 同步发送对象，等待至数据写完，才返回
     * 
     * @param task
     * @param o
     */
    public static void writeSync(Task task, Object o) {
    	synchronized (task) {
			IoSession session = getCurrentSession(task);
			if (session != null) {
				WriteFuture f = session.write(o);
				f.join();
			}
		}
    }
	
	/**
	 * 未决等待
	 * 
	 * @param task
	 */
	public static void setPendingWait(Task task) {
		task.setAttribute(TASK_WAIT_PENDING, "true");	
	}
	
	/**
	 * 跳出未决等待状态
	 * 
	 * @param task
	 * @return
	 */
	public static boolean removePendingWait(Task task) {
		return task.removeAttribute(TASK_WAIT_PENDING) != null ? true : false;
	}
	
	/**
	 * 是否未决等待状态
	 * 
	 * @param task
	 * @return
	 */
	public static boolean isPendingWait(Task task) {
		return task.getAttribute(TASK_WAIT_PENDING) != null ? true : false;
	}
	
	/**
	 * 设置Socket连接对象
	 * 
	 * @param task
	 * @param session
	 */
    public static void setBindSession(Task task, IoSession session) {
    	synchronized (task) {
			if (task != null && session != null) {
				task.setAttribute(FrameworkConstants.TASK_ATTRIBUTE_SESSION,
						session);
			}
		}
    }
    
    /**
     * 获取Task上的有效连接
     * 
     * @param task
     * @return
     */
    public static IoSession getBindSession(Task task) {
    	synchronized (task) {
			return (IoSession) task
					.getAttribute(FrameworkConstants.TASK_ATTRIBUTE_SESSION);
		}
    }
    
    /**
     * 获取Task上的有效连接
     * 
     * @param task
     * @return
     */
    public static IoSession removeBindSession(Task task) {
    	synchronized (task) {
			IoSession session = (IoSession) task
					.removeAttribute(FrameworkConstants.TASK_ATTRIBUTE_SESSION);
			if (session == null || session.isClosing()) {
				return null;
			}

			return session;
		}
    }
    
    /**
	 * 设置Socket连接对象
	 * 
	 * @param task
	 * @param session
	 */
    public static void setCurrentSession(Task task, IoSession session) {
    	synchronized (task) {
			if (task != null && session != null) {
				task.setAttribute(TASK_CURRENT_SESSION, session);
				task.setSession(session);
			}
		}
    }
    
    /**
     * 获取Task上的有效连接
     * 
     * @param task
     * @return
     */
    public static IoSession getCurrentSession(Task task) {
    	synchronized (task) {
			IoSession session = (IoSession) task
					.getAttribute(TASK_CURRENT_SESSION);
			if (session == null) {
				throw new RuntimeException(
						"Can't find CURRENT SESSION. session = " + session);
			}
			if (session.isClosing()) {
				throw new RuntimeException("SESSION CLOSED. session = "
						+ session);
			}

			return session;
		}
    }
    
    /**
     * 获取Task上的有效连接
     * 
     * @param task
     * @return
     */
    public static IoSession removeCurrentSession(Task task) {
    	synchronized (task) {
			IoSession session = (IoSession) task
					.removeAttribute(TASK_CURRENT_SESSION);
			task.setSession(null);
			if (session == null || session.isClosing()) {
				return null;
			}

			return session;
		}
    }
    
    /**
     * 创建或者获取指定Task的同步队列, 如果不存在，则新建一个, 如果存在，则直接返回.
     * 
     * @param task
     * @return
     */
    @SuppressWarnings("unchecked")
	public static ConcurrentTaskQueue<Task> createOrGetSyncCTQFor(Task task, String CTQName) {
    	synchronized (task) {
    		CTQName = TASK_CTQ_ATTR + CTQName;
    		ConcurrentTaskQueue<Task> ctq = (ConcurrentTaskQueue<Task>)task.getAttribute(CTQName);
        	if (ctq == null) {
        		ctq = new ConcurrentTaskQueue<Task>();
        		task.setAttribute(CTQName, ctq);
        	}
        	
        	return ctq;
		}
    }
    
    /**
     * 创建或者获取指定Task的应用服务队列, 如果不存在，则新建一个, 如果存在，则直接返回.
     * 
     * @param task
     * @return
     */
    @SuppressWarnings("unchecked")
	public static ConcurrentTaskQueue<Object> createOrGetServiceCTQFor(Task task) {
    	synchronized (task) {
    		ConcurrentTaskQueue<Object> ctq = (ConcurrentTaskQueue<Object>)task.getAttribute(TASK_SERVICE_CTQ_ATTR);
        	if (ctq == null) {
        		ctq = new ConcurrentTaskQueue<Object>();
        		task.setAttribute(TASK_SERVICE_CTQ_ATTR, ctq);
        	}
        	
        	return ctq;
		}
    }
    
    /**
     * 阻塞等待锁
     * 
     * @param task
     * @return
     */
    private static Object createOrGetWaitTimeoutLock(Task task) {
    	synchronized (task) {
    		Object lock = task.getAttribute(TASK_WAIT_DIRECT_LOCK);
        	if (lock == null) {
        		lock = new Object();
        		task.setAttribute(TASK_WAIT_DIRECT_LOCK, lock);
        	}
        	
        	return lock;
		}    	
    }
    
    /**
     * 当前线程直接在task上阻塞等待
     * 
     * @param task
     */
    public static void waitTimeoutDirect(Task task) {
    	Object lock = createOrGetWaitTimeoutLock(task);
		try {
			TaskUtil.setTimeoutWaitState(task);
			synchronized (lock) {
				try {
					lock.wait(task.getWaitTime());
				} catch (InterruptedException e) {
				}
			}
		} finally {
			TaskUtil.removeTimeoutWaitState(task);
		}
	}
    
    /**
	 * 唤醒在指定Task上阻塞等待的某个线程
	 * 
	 * @param task
	 */
    public static void notifyTaskWaiter(Task task) {
    	synchronized (task) {
			removePendingWait(task);
			Object lock = createOrGetWaitTimeoutLock(task);
			synchronized (lock) {
				lock.notifyAll();

				TaskWaitTimeConsumePool twtcp = getWaitTimePool(task);
				if (twtcp != null) {
					twtcp.removeTaskWaitAndExecute(task);
				}
			}
		}
    }
    
    /**
     * Task失效
     * 
     * @param task
     */
    public static void setTaskStateInValid(Task task) { 
    	task.setAttribute(TASK_STATE_INVALID, "true");
    }
        
    /**
     * 是否失效
     * 
     * @param task
     * @return
     */
    public static boolean isTaskStateInValid(Task task) {
    	return task.getAttribute(TASK_STATE_INVALID) != null ? true : false;
    }
    
    /**
     * task 为等待池中对象
     * 
     * @param task
     */
    public static void setWaitTimePool(Task task, TaskWaitTimeConsumePool pool) {
    	task.setAttribute(TASK_WAIT_TIME_POOL, pool);
    }
    
    /**
     * 获取等待池
     * 
     * @param task
     * @return
     */
    public static TaskWaitTimeConsumePool getWaitTimePool(Task task) {
    	return (TaskWaitTimeConsumePool) task
		                .getAttribute(TASK_WAIT_TIME_POOL);
    }
    
    /**
     * 删除等待池
     * 
     * @param task
     * @return
     */
    public static TaskWaitTimeConsumePool removeWaitTimePool(Task task) {
    	return (TaskWaitTimeConsumePool)task.removeAttribute(TASK_WAIT_TIME_POOL);
    }
    
    /**
     * 是否是等待池中对象
     * 
     * @param task
     * @return
     */
    public static boolean isWaitTimeInPool(Task task) {
    	return task.getAttribute(TASK_WAIT_TIME_POOL) != null ? true : false;
    }
    
    /**
     * 相比前一次，是否又设置了again
     * 
     * @param task
     * @param beforeAgainCount
     * @return
     */
    public static boolean hasSetAgainFor(Task task, int beforeAgainCount) {
    	synchronized (task) {
    	    return beforeAgainCount < task.getAgainCount() && task.isAgain();
    	}
    }
    
    /**
     * 相比前一次，是否又设置了again
     * 
     * @param task
     * @param beforeAgainCount
     * @return
     */
    public static boolean isGenerateResultForCompareWithBefore(Task task, int beforeGenerateResultCount, boolean beforeGenerateResult) {
    	synchronized (task) {
    	    if (task.getGenerateResultCount() <= beforeGenerateResultCount) {
    	    	// 没有设置
    	    	return true;
    	    }
    	
    	    return task.isGenerateResult();
    	}
    }
    
    /**
     * 设置正在超时等待状态
     * 
     * @param task
     */
    public static void setTimeoutWaitState(Task task) {
    	task.setAttribute(TASK_WAIT_TIME_WAIT_STATE, "true");
    }
    
    /**
     * 删除正在超时等待状态
     * 
     * @param task
     * @return
     */
    public static void removeTimeoutWaitState(Task task) {
    	task.removeAttribute(TASK_WAIT_TIME_WAIT_STATE);
    }
    
    /**
     * 是否正在超时等待状态
     * 
     * @param task
     * @return
     */
    public static boolean isTimeoutWaitState(Task task) {
    	return task.getAttribute(TASK_WAIT_TIME_WAIT_STATE) != null ? true : false;
    }
    
    /**
     * 创建或者获取一个计数器
     * 
     * @param task
     * @return
     */
    public static AtomicInteger createOrGetCounterInt(Task task, String counterName) {
    	synchronized (task) {
			counterName = TASK_COUNTER + counterName;
			AtomicInteger c = (AtomicInteger) task.getAttribute(counterName);
			if (c == null) {
				c = new AtomicInteger(0);
				task.setAttribute(counterName, c);
			}

			return c;
		}
    }
    
    /**
	 * 创建或者获取一个计数器
	 * 
	 * @param task
	 * @return
	 */
    public static AtomicLong createOrGetCounterLong(Task task, String counterName) {
    	synchronized (task) {
			counterName = TASK_COUNTER + counterName;
			AtomicLong c = (AtomicLong) task.getAttribute(counterName);
			if (c == null) {
				c = new AtomicLong(0);
				task.setAttribute(counterName, c);
			}

			return c;
		}
    }
}
